package com.atguigu.source;

import com.atguigu.been.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @author wky
 * @create 2021-07-13-16:37
 */

//自定义source
    //需要实现 SourceFunction 接口
    //   如果希望 Source可以指定并行度，那么就 实现 ParallelSourceFunction 这个接口
public class Source_Custom2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.通过自定义数据源获取数据
       senv.addSource(new MySource2("hadoop102",9998)).print();
       senv.execute();

    }

    //TODO 自定义source 
    public static class MySource2 implements SourceFunction<WaterSensor>{
        private String host;
        private int port;
        private java.net.Socket socket;
        private volatile boolean  bool = true ;
        public MySource2(String host,int port){
            this.host=host;
            this.port=port;
        }

        @Override
        public void run(SourceContext<WaterSensor> sourceContext) throws Exception {

            socket = new Socket(host,port);
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            String line = null;

            while (bool && (line=bufferedReader.readLine())!=null){
                String[] words = line.split(" ");
                sourceContext.collect(new WaterSensor(words[0],Long.parseLong(words[1]),Integer.parseInt(words[2])));
            }
        }

        @Override
        public void cancel() {
            bool=false;
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

